package com.mq.demo.listener;

import lombok.extern.java.Log;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQPushConsumerLifecycleListener;
import org.springframework.stereotype.Component;

import java.util.List;

/**
 * @Author Re
 * @create 2023/2/28 16:51
 * @Description 消费者处理 push模式 RocketMQPushConsumerLifecycleListener优先于RocketMQListener  取其一消费即可  这里测试用
 */
@Component
@RocketMQMessageListener(consumerGroup = "${rocketmq.producer.imConsumerGroup}", topic = "${rocketmq.producer.imToCommonMsg}")
@Log
public class RocketConsumeListener implements RocketMQListener<MessageExt>, RocketMQPushConsumerLifecycleListener{

    @Override
    public void onMessage(MessageExt messageExt) {
        log.info("用户微服务收到了mq信息：" + messageExt);
    }

    @Override
    public void prepareStart(DefaultMQPushConsumer consumer) {
        /**
         * 1. CONSUME_FROM_LAST_OFFSET：第一次启动从队列最后位置消费，后续再启动接着上次消费的进度开始消费
         * 2. CONSUME_FROM_FIRST_OFFSET：第一次启动从队列初始位置消费，后续再启动接着上次消费的进度开始消费
         * 3. CONSUME_FROM_TIMESTAMP：第一次启动从指定时间点位置消费，后续再启动接着上次消费的进度开始消费
         * 以上所说的第一次启动是指从来没有消费过的消费者，如果该消费者消费过，那么会在broker端记录该消费者的消费位置，如果该消费者挂了再启动，那么自动从上次消费的进度开始
         */
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        /**
         * CLUSTERING(集群模式)：默认模式，同一个消费组内的消费者分担消费，从而达到负载均衡的目的
         * BROADCASTING(广播模式)：广播模式下，消费组内的每一个消费者都会消费全量消息
         */
        consumer.setMessageModel(MessageModel.CLUSTERING);
        //注册回调接口来处理从Broker中收到的消息
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> mgs, ConsumeConcurrentlyContext context) {
                log.info(Thread.currentThread().getName()+"Receive New Messages:"+mgs);
                for (MessageExt msg:mgs) {
                    messageProcessor(msg);
                }
                /**
                 * 并发消费返回消息消费状态:
                 * ConsumeConcurrentlyStatus.CONSUME_SUCCESS为消费成功
                 * ConsumeConcurrentlyStatus.RECONSUME_LATER boker会根据设置的messageDelayLevel发起重试，默认16次
                 * 顺序消费返回消息消费状态:
                 * ConsumeOrderlyStatus.SUCCESS表示消费成功
                 * ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT表示消费失败
                 */
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
    }

    /**
     * 消息处理
     * @param msg 消息参数
     * @return
     */
    public void messageProcessor(MessageExt msg){
        String str = new String(msg.getBody());
        log.info("消息内容：" + str);
    }
}
